
package com.echo.sdk.im.server;

import com.alibaba.fastjson.JSON;
import com.echo.sdk.im.handler.CIMRequestHandler;
import com.echo.sdk.im.constant.CIMConstant;
import com.echo.sdk.im.handler.HeartbeatHandler;
import com.echo.sdk.im.model.CIMMsg;
import com.echo.sdk.im.model.CIMSession;
import com.echo.sdk.im.model.HeartbeatRequest;
import com.echo.sdk.im.model.HeartbeatResponse;
import com.echo.sdk.im.util.MessageUtils;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.http.FullHttpRequest;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpServerCodec;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.stream.ChunkedWriteHandler;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.AttributeKey;
import lombok.extern.slf4j.Slf4j;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.HashMap;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;


/**
 * @author chentl
 * @version V1.0
 * @Project IM-SDK
 * @Title CIMNioSocketServer.java
 * @Description 自定义IM服务器配置类
 * @Package com.whdsj.sdk.im.server.handler
 * @date 2020/9/6 4:51 下午
 */
@Slf4j
public class CIMNioSocketServer {

    private static final Logger LOGGER = LoggerFactory.getLogger(CIMNioSocketServer.class);

    /**
     * 读空闲时间(秒)
     */
    public static final int READ_IDLE_TIME = 30;

    /**
     * 写接空闲时间(秒)
     */
    public static final int WRITE_IDLE_TIME = 30;

    /**
     * 心跳响应 超时为15秒
     */
    public static final int PONG_TIME_OUT = 15;

    //内置特殊处理handler
    private final HashMap<String, CIMRequestHandler> innerHandlerMap = new HashMap<>();


    /**
     * @description 线程安全的所有频道组 @author chentl @date 2020/9/7 3:42 下午
     **/
    private final ConcurrentHashMap<String, Channel> channelGroup = new ConcurrentHashMap<>();

    /**
     * @description 存放所有user对应的所有 channels @author chentl @date 2020/9/7 3:44 下午
     **/
//    public final static Map<String, ConcurrentHashSet<String>> globalUserChannels = new ConcurrentHashMap<>();

    private EventLoopGroup webBossGroup;
    private EventLoopGroup webWorkerGroup;

    private final String webHost;
    private final Integer webPort;

    private final CIMRequestHandler outerRequestHandler;

    private final ChannelHandler defaultChannelEventHandler = new CommonChannelEventHandler();

    /**
     * @return
     * @author chentl
     * @description 绑定构造
     * @date 2020/9/6 2:59 下午
     * @params [builder]
     */
    public CIMNioSocketServer(Builder builder) {
        this.webPort = builder.webPort;
        this.webHost = builder.webHost;
        this.outerRequestHandler = builder.outerRequestHandler;
    }


    /**
     * @author chentl
     * @description 绑定服务构造实现类
     * @date 2020/9/6 2:58 下午
     * @params
     * @return
     */
    public static class Builder {

        private String webHost;
        private Integer webPort;
        private CIMRequestHandler outerRequestHandler;

        public Builder setWebsocketConfig(String host, Integer port) {
            this.webHost = host;
            this.webPort = port;
            return this;
        }

        /**
         * @return com.whdsj.sdk.im.server.handler.CIMNioSocketServer.Builder
         * @author chentl
         * @description 设置应用层的消息处理handler
         * @date 2020/9/6 2:59 下午
         * @params [outerRequestHandler]
         */
        public Builder setOuterRequestHandler(CIMRequestHandler outerRequestHandler) {
            this.outerRequestHandler = outerRequestHandler;
            return this;
        }

        /**
         * @return com.whdsj.sdk.im.server.handler.CIMNioSocketServer
         * @author chentl
         * @description 绑定函数
         * @date 2020/9/6 3:00 下午
         * @params []
         */
        public CIMNioSocketServer build() {
            return new CIMNioSocketServer(this);
        }

    }

    /**
     * @return void
     * @author chentl
     * @description Netty服务绑定WEB端口
     * @date 2020/9/6 3:00 下午
     * @params []
     */
    public void bind() {
        innerHandlerMap.put(CIMConstant.CLIENT_HEARTBEAT, new HeartbeatHandler());

        if (webPort != null) {
            bindWebPort();
        }
    }

    /**
     * @return void
     * @author chentl
     * @description 绑定web端口
     * @date 2020/9/6 3:04 下午
     * @params []
     */
    private void bindWebPort() {
        webBossGroup = new NioEventLoopGroup();
        webWorkerGroup = new NioEventLoopGroup();
        ServerBootstrap bootstrap = createServerBootstrap(webBossGroup, webWorkerGroup);
        bootstrap.childHandler(new ChannelInitializer<SocketChannel>() {

            @Override
            public void initChannel(SocketChannel ch) {
                ChannelPipeline pipeline = ch.pipeline();
                pipeline.addLast(new HttpServerCodec());
                pipeline.addLast(new ChunkedWriteHandler());
                pipeline.addLast(new HttpObjectAggregator(65536));
                pipeline.addLast(new WebSocketServerProtocolHandler("/ws"));
                pipeline.addLast(new IdleStateHandler(READ_IDLE_TIME, WRITE_IDLE_TIME, 0, TimeUnit.SECONDS));
                pipeline.addLast(new LoggingHandler(LogLevel.INFO));
                pipeline.addLast(defaultChannelEventHandler);
            }

        });

        ChannelFuture channelFuture = bootstrap.bind(webHost, webPort).syncUninterruptibly();
        channelFuture.channel().newSucceededFuture().addListener(future -> {
            String logBanner = "\n\n" +
                    "* * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * *\n" +
                    "*                                                                                   *\n" +
                    "*                                                                                   *\n" +
                    "*                   Websocket Server started on host {} port {}.                         *\n" +
                    "*                                                                                   *\n" +
                    "*                                                                                   *\n" +
                    "* * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * *\n";
            LOGGER.info(logBanner, webHost, webPort);
        });
        channelFuture.channel().closeFuture().addListener(future -> this.destroy(webBossGroup, webWorkerGroup));
    }

    /**
     * @return io.netty.bootstrap.ServerBootstrap
     * @author chentl
     * @description 初始化服务参数
     * @date 2020/9/6 3:01 下午
     * @params [bossGroup, workerGroup]
     */
    private ServerBootstrap createServerBootstrap(EventLoopGroup bossGroup, EventLoopGroup workerGroup) {
        ServerBootstrap bootstrap = new ServerBootstrap();
        bootstrap.option(ChannelOption.SO_BACKLOG, CIMConstant.SO_BACKLOG)
                .group(bossGroup, workerGroup)
                .channel(NioServerSocketChannel.class)
                .childOption(ChannelOption.TCP_NODELAY, true)
                .childOption(ChannelOption.SO_KEEPALIVE, true);

        return bootstrap;
    }


    /**
     * @return void
     * @author chentl
     * @description 销毁主从线程接收对象, 释放资源
     * @date 2020/9/6 3:03 下午
     * @params [bossGroup, workerGroup]
     */
    public void destroy(EventLoopGroup bossGroup, EventLoopGroup workerGroup) {
        if (bossGroup != null && !bossGroup.isShuttingDown() && !bossGroup.isShutdown()) {
            try {
                bossGroup.shutdownGracefully();
            } catch (Exception ignore) {
            }
        }

        if (workerGroup != null && !workerGroup.isShuttingDown() && !workerGroup.isShutdown()) {
            try {
                workerGroup.shutdownGracefully();
            } catch (Exception ignore) {
            }
        }
    }

    /**
     * 关闭长连接服务
     */
    public void destroy() {
        this.destroy(webBossGroup, webWorkerGroup);
    }

    /**
     * @return io.netty.channel.Channel
     * @author chentl
     * @description 获取当前Channel
     * @date 2020/9/6 3:49 下午
     * @params [id]
     */
    public Channel getManagedSession(String id) {
        if (id == null) {
            return null;
        }
        return channelGroup.get(id);
    }


    @Sharable
    private class CommonChannelEventHandler extends SimpleChannelInboundHandler<Object> {
        /**
         * @return void
         * @author chentl
         * @description 消息入口
         * @date 2020/9/6 4:49 下午
         * @params [ctx, data]
         */
        @Override
        protected void channelRead0(ChannelHandlerContext ctx, Object data) {
            //心跳响应忽略
            if (data instanceof HeartbeatResponse) {
                return;
            }
            CIMSession session = new CIMSession(ctx.channel());
            TextWebSocketFrame msg = (TextWebSocketFrame) data;

            CIMMsg body = MessageUtils.convertMsg(msg.text());
            if (Objects.isNull(body)) {
                log.error("[cim] 消息格式异常，无法解析。原始消息:{}", JSON.toJSONString(body));
                return;
            }

            CIMRequestHandler handler = innerHandlerMap.get(body.getType());

            if (handler != null) {
                handler.process(session, body);
                return;
            }
            log.info("[server]收到客户端消息:{}", msg.text());
            outerRequestHandler.process(session, body);

            //TODO 传统的http接入 后续扩展
            if (data instanceof FullHttpRequest) {
                log.info("传统的http消息接入");
            }
        }

        /**
         * @return void
         * @author chentl
         * @description Channel失活
         * @date 2020/9/6 4:50 下午
         * @params [ctx]
         */
        @Override
        public void channelInactive(ChannelHandlerContext ctx) {
            channelGroup.remove(ctx.channel().id().asShortText());

            CIMSession session = new CIMSession(ctx.channel());
            CIMMsg cimMsg = new CIMMsg();
            cimMsg.setType(CIMConstant.CLIENT_CONNECT_CLOSED);
            outerRequestHandler.process(session, cimMsg);

        }

        /**
         * @return void
         * @author chentl
         * @description Channel激活
         * @date 2020/9/6 4:50 下午
         * @params [ctx]
         */
        @Override
        public void channelActive(ChannelHandlerContext ctx) {
            log.info("[server]新连接 {} 加入!", ctx.channel().id());
            channelGroup.put(ctx.channel().id().asShortText(), ctx.channel());
            log.info("[server]当前在线人数:" + channelGroup.size());
        }

        /**
         * @return void
         * @author chentl
         * @description 加入新连接时触发
         * @date 2020/1/2 14:39
         * @params [ctx]
         * @since JDK 1.7
         */
//        @Override
//        public void handlerAdded(ChannelHandlerContext ctx) {
//            log.info("[server]新的");
//        }

        /**
         * @return void
         * @author chentl
         * @description 连接断开
         * @date 2020/1/2 14:40
         * @params [ctx]
         * @since JDK 1.7
         */
        @Override
        public void handlerRemoved(ChannelHandlerContext ctx) {
            log.info("[server]连接 {} 现已断开", ctx.channel().id());
            //TODO 从房间中移除一个连接时触发
            log.info("[server]当前在线人数:" + channelGroup.size());

        }

        /**
         * @return void
         * @author chentl
         * @description 连接异常时需要及时关闭相关资源
         * @date 2020/1/2 14:48
         * @params [ctx, cause]
         * @since JDK 1.7
         */
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
            log.error("[server]连接websocket异常" + cause.toString());
            ctx.close();
            ctx.channel().close();
        }

        /**
         * @return void
         * @author chentl
         * @description 挤一挤管道中的消息
         * @date 2020/1/2 14:48
         * @params [ctx]
         * @since JDK 1.7
         */
        @Override
        public void channelReadComplete(ChannelHandlerContext ctx) {
            ctx.flush();
        }

        /**
         * @return void
         * @author chentl
         * @description Socket事件触发
         * @date 2020/9/6 4:51 下午
         * @params [ctx, evt]
         */
        @Override
        public void userEventTriggered(ChannelHandlerContext ctx, Object evt) {

            if (!(evt instanceof IdleStateEvent)) {
                return;
            }

            IdleStateEvent idleEvent = (IdleStateEvent) evt;

            if (idleEvent.state().equals(IdleState.WRITER_IDLE)) {
                String heartPack = HeartbeatRequest.getInstance().toJsonString();
                ctx.channel().attr(AttributeKey.valueOf(CIMConstant.HEARTBEAT_KEY)).set(System.currentTimeMillis());
                ctx.channel().writeAndFlush(new TextWebSocketFrame(heartPack));
                log.info("[server nId:{}]发送服务端心跳:{} ", ctx.channel().id(), HeartbeatRequest.getInstance().getContent());
            }

            /*
             * 如果心跳请求发出10秒内没收到响应，则关闭连接
             */
            if (idleEvent.state().equals(IdleState.READER_IDLE)) {

                Long lastTime = (Long) ctx.channel().attr(AttributeKey.valueOf(CIMConstant.HEARTBEAT_KEY)).get();
                if (lastTime != null && System.currentTimeMillis() - lastTime >= PONG_TIME_OUT) {
                    ctx.channel().close();
                }

                ctx.channel().attr(AttributeKey.valueOf(CIMConstant.HEARTBEAT_KEY)).set(null);
            }
        }

    }

}
